package dongshi.daddy.paxos.basicpaxos;

import cn.hutool.json.JSONUtil;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

// basic paxos
public class BasicPaxosTest {
    private static CountDownLatch allNodeReady = new CountDownLatch(5);
    private static CountDownLatch mainNotExits = new CountDownLatch(5);
    private static String CLIENT_1_NAME = "client1";
    private static int CLIENT_1_PORT = 6666;
    private static String CLIENT_2_NAME = "client2";
    private static int CLIENT_2_PORT = 6667;
    private static String NODE_A_NAME = "nodeA";
    private static int NODE_A_PORT = 6668;
    private static String NODE_B_NAME = "nodeB";
    private static int NODE_B_PORT = 6669;
    private static String NODE_C_NAME = "nodeC";
    private static int NODE_C_PORT = 6670;
    private static String PREPARED_PHASE = "(准备阶段)";
    private static String ACCEPT_PHASE = "(接受阶段)";
    private static String NO_PROPOSAL = "[尚无提案]";
    private static Map<String, Integer> CLIENT_PORT_MAP = new HashMap() {
        {
            put(CLIENT_1_NAME, CLIENT_1_PORT);
            put(CLIENT_2_NAME, CLIENT_2_PORT);
            put(NODE_A_NAME, NODE_A_PORT);
            put(NODE_B_NAME, NODE_B_PORT);
            put(NODE_C_NAME, NODE_C_PORT);
        }
    };

    public static void main(String[] args) throws Exception {
        Thread client1 = new Thread(() -> {
            try {
                String whoAmI = CLIENT_1_NAME;
                ServerSocket client1Socket = new ServerSocket(CLIENT_1_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                // 发送准备阶段消息给节点A,节点B
                Proposal proposalToAAndB = new Proposal();
                proposalToAAndB.setWhoAmI(whoAmI);
                proposalToAAndB.setWhoAmI(CLIENT_1_NAME);
                proposalToAAndB.setSerialNum(1000);
//                proposalToAAndB.setVal("2");
                sendMsg(JSONUtil.toJsonStr(proposalToAAndB), NODE_A_PORT);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndB), NODE_B_PORT);
                // 准备接收来自节点A和节点B的响应
                new Thread(() -> {
                    int counter = 0;
                    while (counter < 2) {
                        counter++;
                        try {
                            String msg = receiveMsg(client1Socket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到回复消息：" + msg);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                // 稍等一会，发送消息给节点C
                Thread.sleep(300);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndB), NODE_C_PORT);
                // 准备接收来自节点C的响应(这里是收不到响应的，因为节点C的最小提案编号是2000，因此自己发送的消息会被丢弃)
                new Thread(() -> {
                    int counter = 0;
                    while (counter < 2) {
                        counter++;
                        try {
                            String msg = receiveMsg(client1Socket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到回复消息：" + msg);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                // 稍等一会，进入接受阶段，发送消息给节点A,节点B，节点C
                Thread.sleep(300);
                Proposal proposalToAAndBInCommit = new Proposal();
                proposalToAAndBInCommit.setWhoAmI(whoAmI);
                proposalToAAndBInCommit.setWhoAmI(CLIENT_1_NAME);
                proposalToAAndBInCommit.setSerialNum(1000);
                proposalToAAndBInCommit.setVal("2");
                sendMsg(JSONUtil.toJsonStr(proposalToAAndBInCommit), NODE_A_PORT);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndBInCommit), NODE_B_PORT);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndBInCommit), NODE_C_PORT);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "client1");
        client1.start();

        Thread client2 = new Thread(() -> {
            try {
                String whoAmI = CLIENT_2_NAME;
                ServerSocket client2Socket = new ServerSocket(CLIENT_2_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                // 发送准备阶段消息给节点C
                Proposal proposalToAAndB = new Proposal();
                proposalToAAndB.setWhoAmI(whoAmI);
                proposalToAAndB.setSerialNum(2000);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndB), NODE_C_PORT);
                // 准备接收来自节点C的响应
                new Thread(() -> {
                    int counter = 0;
                    while (counter < 1) {
                        counter++;
                        try {
                            String msg = receiveMsg(client2Socket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到回复消息：" + msg);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                // 稍等一会，发送消息给节点A和节点B
                Thread.sleep(300);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndB), NODE_A_PORT);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndB), NODE_B_PORT);
                // 准备接收来自节点A和节点B的响应
                new Thread(() -> {
                    int counter = 0;
                    while (counter < 2) {
                        counter++;
                        try {
                            String msg = receiveMsg(client2Socket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到回复消息：" + msg);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                // 稍等一会，进入接受阶段，发送消息给节点A,节点B，节点C
                Thread.sleep(300);
                Proposal proposalToAAndBInCommit = new Proposal();
                proposalToAAndBInCommit.setWhoAmI(whoAmI);
                proposalToAAndBInCommit.setWhoAmI(CLIENT_2_NAME);
                proposalToAAndBInCommit.setSerialNum(2000);
                proposalToAAndBInCommit.setVal("7");
                sendMsg(JSONUtil.toJsonStr(proposalToAAndBInCommit), NODE_A_PORT);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndBInCommit), NODE_B_PORT);
                sendMsg(JSONUtil.toJsonStr(proposalToAAndBInCommit), NODE_C_PORT);
            } catch (Exception e) {
            }
        }, "client2");
        client2.start();

        Thread nodeA = new Thread(() -> {
            try {
                String whoAmI = NODE_A_NAME;
                ServerSocket nodeASocket = new ServerSocket(NODE_A_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                new Thread(() -> {
                    boolean preparePhaseFinish = false;
                    List<String> preparePhaseRecMsgList = new ArrayList<>();
                    // 准备阶段会进行应答的最小的提案编号
                    int allowMinSerialNum = -1;
                    while (!preparePhaseFinish) {
                        try {
                            String msg = receiveMsg(nodeASocket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到消息：" + msg);
                            preparePhaseRecMsgList.add(msg);
                            Proposal receiveProposal = JSONUtil.toBean(msg, Proposal.class);
                            int receiveSerialNum = receiveProposal.getSerialNum();
                            // 根据当前拥有的最小提案编号判断是给予回复，还是直接丢弃
                            if (allowMinSerialNum <= receiveSerialNum) {
                                allowMinSerialNum = receiveSerialNum;
                                Res res = new Res();
                                res.setWhoAmI(whoAmI);
                                res.setContent(NO_PROPOSAL);
                                sendMsg(JSONUtil.toJsonStr(res), CLIENT_PORT_MAP.get(receiveProposal.getWhoAmI()));
                            } else {
                                System.out.println(PREPARED_PHASE + whoAmI + "收到消息：" + msg + "（丢弃）");
                            }
                            preparePhaseFinish = (preparePhaseRecMsgList.size() >= 2);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    boolean acceptPhaseFinish = false;
                    List<String> acceptPhaseRecMsgList = new ArrayList<>();
                    // 准备接收接受阶段消息
                    while (!acceptPhaseFinish) {
                        try {
                            String msg = receiveMsg(nodeASocket);
                            System.out.println(ACCEPT_PHASE + whoAmI + "收到消息：" + msg);
                            acceptPhaseRecMsgList.add(msg);
                            Proposal receiveProposal = JSONUtil.toBean(msg, Proposal.class);
                            int receiveSerialNum = receiveProposal.getSerialNum();
                            // 根据当前拥有的最小提案编号判断是给予回复，还是直接丢弃
                            if (allowMinSerialNum <= receiveSerialNum) {
                                System.out.println(ACCEPT_PHASE + whoAmI + "达成共识：" + msg);
                            } else {
                                System.out.println(ACCEPT_PHASE + whoAmI + "收到消息：" + msg + "（丢弃）");
                            }
                            acceptPhaseFinish = (acceptPhaseRecMsgList.size() >= 2);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "nodeA");
        nodeA.start();

        Thread nodeB = new Thread(() -> {
            try {
                String whoAmI = NODE_B_NAME;
                ServerSocket nodeBSocket = new ServerSocket(NODE_B_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                new Thread(() -> {
                    boolean preparePhaseFinish = false;
                    List<String> preparePhaseRecMsgList = new ArrayList<>();
                    // 准备阶段会进行应答的最小的提案编号
                    int allowMinSerialNum = -1;
                    while (!preparePhaseFinish) {
                        try {
                            String msg = receiveMsg(nodeBSocket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到消息：" + msg);
                            preparePhaseRecMsgList.add(msg);
                            Proposal receiveProposal = JSONUtil.toBean(msg, Proposal.class);
                            int receiveSerialNum = receiveProposal.getSerialNum();
                            // 根据当前拥有的最小提案编号判断是给予回复，还是直接丢弃
                            if (allowMinSerialNum <= receiveSerialNum) {
                                allowMinSerialNum = receiveSerialNum;
                                Res res = new Res();
                                res.setWhoAmI(whoAmI);
                                res.setContent(NO_PROPOSAL);
                                sendMsg(JSONUtil.toJsonStr(res), CLIENT_PORT_MAP.get(receiveProposal.getWhoAmI()));
                            } else {
                                System.out.println(PREPARED_PHASE + whoAmI + "收到消息：" + msg + "（丢弃）");
                            }
                            preparePhaseFinish = (preparePhaseRecMsgList.size() >= 2);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    boolean acceptPhaseFinish = false;
                    List<String> acceptPhaseRecMsgList = new ArrayList<>();
                    // 准备接收接受阶段消息
                    while (!acceptPhaseFinish) {
                        try {
                            String msg = receiveMsg(nodeBSocket);
                            System.out.println(ACCEPT_PHASE + whoAmI + "收到消息：" + msg);
                            acceptPhaseRecMsgList.add(msg);
                            Proposal receiveProposal = JSONUtil.toBean(msg, Proposal.class);
                            int receiveSerialNum = receiveProposal.getSerialNum();
                            // 根据当前拥有的最小提案编号判断是给予回复，还是直接丢弃
                            if (allowMinSerialNum <= receiveSerialNum) {
                                System.out.println(ACCEPT_PHASE + whoAmI + "达成共识：" + msg);
                            } else {
                                System.out.println(ACCEPT_PHASE + whoAmI + "收到消息：" + msg + "（丢弃）");
                            }
                            acceptPhaseFinish = (acceptPhaseRecMsgList.size() >= 2);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "nodeB");
        nodeB.start();

        Thread nodeC = new Thread(() -> {
            try {
                String whoAmI = NODE_C_NAME;
                ServerSocket nodeCSocket = new ServerSocket(NODE_C_PORT);
                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                new Thread(() -> {
                    boolean preparePhaseFinish = false;
                    List<String> preparePhaseRecMsgList = new ArrayList<>();
                    // 准备阶段会进行应答的最小的提案编号
                    int allowMinSerialNum = -1;
                    while (!preparePhaseFinish) {
                        try {
                            String msg = receiveMsg(nodeCSocket);
                            System.out.println(PREPARED_PHASE + whoAmI + "收到消息：" + msg);
                            preparePhaseRecMsgList.add(msg);
                            Proposal receiveProposal = JSONUtil.toBean(msg, Proposal.class);
                            int receiveSerialNum = receiveProposal.getSerialNum();
                            // 根据当前拥有的最小提案编号判断是给予回复，还是直接丢弃
                            if (allowMinSerialNum <= receiveSerialNum) {
                                allowMinSerialNum = receiveSerialNum;
                                Res res = new Res();
                                res.setWhoAmI(whoAmI);
                                res.setContent(NO_PROPOSAL);
                                sendMsg(JSONUtil.toJsonStr(res), CLIENT_PORT_MAP.get(receiveProposal.getWhoAmI()));
                            } else {
                                System.out.println(PREPARED_PHASE + whoAmI + "收到消息：" + msg + "（丢弃）");
                            }
                            preparePhaseFinish = (preparePhaseRecMsgList.size() >= 2);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        boolean acceptPhaseFinish = false;
                        List<String> acceptPhaseRecMsgList = new ArrayList<>();
                        // 准备接收接受阶段消息
                        while (!acceptPhaseFinish) {
                            try {
                                String msg = receiveMsg(nodeCSocket);
                                System.out.println(ACCEPT_PHASE + whoAmI + "收到消息：" + msg);
                                acceptPhaseRecMsgList.add(msg);
                                Proposal receiveProposal = JSONUtil.toBean(msg, Proposal.class);
                                int receiveSerialNum = receiveProposal.getSerialNum();
                                // 根据当前拥有的最小提案编号判断是给予回复，还是直接丢弃
                                if (allowMinSerialNum <= receiveSerialNum) {
                                    System.out.println(ACCEPT_PHASE + whoAmI + "达成共识：" + msg);
                                } else {
                                    System.out.println(ACCEPT_PHASE + whoAmI + "收到消息：" + msg + "（丢弃）");
                                }
                                acceptPhaseFinish = (acceptPhaseRecMsgList.size() >= 2);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }).start();

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "nodeC");
        nodeC.start();
        mainNotExits.await();
    }


    public static void sendMsg(String sendMsg, int port) throws Exception {
        Socket socket = new Socket();
        // 设置超时不生效，这里手动休眠
        Thread.sleep(200);
        socket.connect(new InetSocketAddress("127.0.0.1", port));
        OutputStream os = socket.getOutputStream();
        PrintWriter pw = new PrintWriter(os);
        pw.println(sendMsg);
        pw.flush();
        socket.close();
    }


    public static String receiveMsg(ServerSocket serverA) throws IOException {
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }
}
